package com.loginfail_detect

import com.loginfail_detect.bean.{LoginEvent, LoginFailWarning}
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector



/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/12/4 20:37
  *                统计2秒钟连续登录失败，并输出
  *                直接判断，不需要等2s,(缺陷：无法处理迟到登录数据)
  */
object LoginFailAdvance {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //定义事件时间语义

    //从文件中读取数据，并转换成样例类,提取时间戳生成watermark
    val filePath: String = "LoginFailDetect/src/main/resources/LoginLog.csv"
    val fileStream: DataStream[String] = env.readTextFile(filePath)

    //转换成样例类类型，并指定时间戳和watermark
    val loginEventStream: DataStream[LoginEvent] = fileStream.map(data => {
      val arr: Array[String] = data.split(",")
      LoginEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
    })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEvent](Time.seconds(3)) {
        override def extractTimestamp(t: LoginEvent): Long = t.timestamp * 1000L
      })
    //进行判断和检测，如果超过2s之内连续登录失败，输出报警信息
    val loginFailWarningStream: DataStream[LoginFailWarning] = loginEventStream.keyBy(_.userId)
      .process(new LoginFailWarningAdvanceResult(2))

    loginFailWarningStream.print("fail")

    env.execute("login fail delete job")

  }

  class LoginFailWarningAdvanceResult(failTimes: Int) extends KeyedProcessFunction[Long, LoginEvent, LoginFailWarning] {
    //定义状态，保存当前所有的登录失败事件，保存定时器的时间戳
    lazy val loginFailListState: ListState[LoginEvent] = getRuntimeContext
      .getListState(new ListStateDescriptor[LoginEvent]("loginfail-list", classOf[LoginEvent]))


    override def processElement(i: LoginEvent, context: KeyedProcessFunction[Long, LoginEvent, LoginFailWarning]#Context, collector: Collector[LoginFailWarning]): Unit = {
      //判断当前登录事件是否成功还是失败
      if (i.eventType == "fail") {
        //如果是失败，进一步做判断
        val iter = loginFailListState.get().iterator()
        //判断之前是否有登录失败事件
        if(iter.hasNext){
          //如果有，那么判断两次失败的时间差
          val firstFailEvent =iter.next()
          if(i.timestamp<firstFailEvent.timestamp+failTimes){
            //如果在2秒内，直接输出报警
            collector.collect(LoginFailWarning(
              i.userId,
              firstFailEvent.timestamp,
              i.timestamp,
            "login fail 2 times in 2s"))
          }
          //不管报不报警，当前都已经处理完毕，将状态更新为最近一次登录失败事件
          loginFailListState.clear()
          loginFailListState.add(i)
        }else{
          //如果没有,直接把当前事件添加到ListSate中
          loginFailListState.add(i)
        }


      } else {
        //如果是成功，直接清空状态
        loginFailListState.clear()
      }
    }

    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, LoginEvent, LoginFailWarning]#OnTimerContext, out: Collector[LoginFailWarning]): Unit = {

    }
  }


}
